package com.jarveis.frame.util;

import java.io.IOException;

import org.apache.log4j.Logger;
import org.zbus.broker.Broker;
import org.zbus.broker.BrokerConfig;
import org.zbus.broker.ZbusBroker;
import org.zbus.mq.Consumer;
import org.zbus.mq.Consumer.ConsumerHandler;
import org.zbus.mq.Producer;
import org.zbus.net.http.Message;
import org.zbus.net.http.Message.MessageInvoker;
import org.zbus.rpc.RpcFactory;
import org.zbus.rpc.mq.MqInvoker;

import com.jarveis.dbs.core.DBService;

/**
 * 数据访问工具类
 * 
 * @author liuguojun
 */
public class DBS4ZBusClient {

	private static final Logger logger = Logger.getLogger(DBS4ZBusClient.class);

	public static final String ERR_CODE = "@errcode";
	public static final String ERR_DESC = "@errdesc";
	
	private static final String MQ = "MQ_"; 
	private static Broker broker = null;

	/**
	 * 开启
	 * 
	 * @param host
	 * @throws IOException
	 */
	public static void start(String host) throws IOException {
		if (broker == null) {
			BrokerConfig config = new BrokerConfig();
			config.setBrokerAddress(host);
			broker = new ZbusBroker(config);
		}
	}

	/**
	 * 关闭
	 * 
	 * @throws IOException
	 */
	public static void stop() throws IOException {
		if (broker != null) {
			broker.close();
		}
	}

	/**
	 * 访问接口
	 * 
	 * @param in
	 * @return
	 */
	public static Param callInterface(Param in) {
		return callPlain(in.toString());
	}

	/**
	 * 明文访问接口
	 * 
	 * @param link
	 * @param encoding
	 * @param inStr
	 * @return
	 */
	private static Param callPlain(String inStr) {
		Param out = null;
		try {
			out = new Param("Resp");
			
			MessageInvoker invoker = new MqInvoker(broker, "DBSRpc");
			RpcFactory factory = new RpcFactory(invoker);

			DBService service = factory.getService(DBService.class);
			String outStr = service.process(inStr);
			out = new Param(outStr);
		} catch (Exception ex) {
			logger.error(ex.getMessage(), ex);
		}

		return out;
	}

	/**
	 * 添加消息到自定义的对列中
	 * 
	 * @param key
	 * @param value
	 * @return
	 */
	public static boolean push(String key, String value) {
		try {
			Producer producer = new Producer(broker, MQ + key);
			producer.createMQ();

			Message msg = new Message();
			msg.setBody(value);
			producer.sendSync(msg);
			
			return true;
		} catch (Exception ex) {
			logger.error(ex.getMessage(), ex);
		}
		
		return false;
	}
	
	/**
	 * 弹出自定义的对列中的消息
	 * 
	 * @param key
	 * @param value
	 * @return
	 */
	public static void pop(String key, ConsumerHandler handler) {
		try {
			Consumer consumer = new Consumer(broker, MQ + key);
			consumer.start(handler);
		} catch (Exception ex) {
			logger.error(ex.getMessage(), ex);
		}
	}
}
